package com.mxnavi5.example.DataStream;

import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.scala.typeutils.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.sql.Array;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.TimeUnit;

public class MyStateTest {

    public static void main(String[] args) throws Exception {
        //    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        Configuration configuration = new Configuration();
        configuration.setInteger(RestOptions.PORT, 8081);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(configuration);
        env.setParallelism(1);
        env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime);

        SingleOutputStreamOperator<Integer> dataStreamSource = env.addSource(new UdfSourceTest2()).returns(Integer.class);



        //process   自定义逻辑
//        SingleOutputStreamOperator<Integer> processStream = dataStreamSource.process(new TestProcessFunction());
//        processStream.print("prcess");

        SingleOutputStreamOperator<Tuple2<Integer, Integer>> mapStream = dataStreamSource.map(x -> Tuple2.of(x%2, x))
                .returns(TypeInformation.of(new TypeHint<Tuple2<Integer, Integer>>() {
                }));

        KeyedStream<Tuple2<Integer, Integer>, Integer> keyStream = mapStream.keyBy(t -> t.f0);


        //KeyedProcessFunction
        keyStream.process(new TestKeyedProcessFunction()).print("keyedProcess");




        env.execute();


    }

}
